package org.apache.flink.connector.jdbc.databases.mysql.dialect;


import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.connector.jdbc.split.JdbcGenericParameterValuesProvider;

import java.util.Arrays;

public class testJdbcInputFormat {

    public static void main(String[] args) {

        TypeInformation<?>[] fieldTypes = new TypeInformation<?>[] {

                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.DOUBLE_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO
        };

        RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes);
        
        String[][] queryParameters = new String[2][1];
        queryParameters[0] = new String[]{"Kumar"};
        queryParameters[1] = new String[]{"Tan Ah Teck"};
        // 打印
        for (int i = 0; i < queryParameters.length; i++) {
            System.out.println(Arrays.toString(queryParameters[i]));
        }

        JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://hadoop105:3306/testdb")
                .setQuery("select * from books WHERE author = ?")
                .setRowTypeInfo(rowTypeInfo)
                .setParametersProvider(new JdbcGenericParameterValuesProvider(queryParameters))
                .finish();

        System.out.println(jdbcInputFormat);
    }
}
